package com.company.wushuo.justfortest.reactive;

import android.support.annotation.NonNull;

import com.company.wushuo.justfortest.reactive.function.Func1;
import com.company.wushuo.justfortest.reactive.internal.operators.OnSubscribeLift;
import com.company.wushuo.justfortest.reactive.internal.operators.OnSubscribeMap;
import com.company.wushuo.justfortest.reactive.internal.operators.Operator;
import com.company.wushuo.justfortest.reactive.internal.operators.OperatorObserveOn;
import com.company.wushuo.justfortest.reactive.subscription.Subscriptions;

import rx.Scheduler;


/**
 * 自己动手写RxJava
 * Created by wushuo on 2021/4/3.
 */

public class Observable<T> implements ObservableSource<T> {

    final OnSubscribe<T> subscribe;


    protected Observable(OnSubscribe<T> onSubscribe) {
        subscribe = onSubscribe;
    }

    public static <T> Observable<T> create(OnSubscribe<T> onSubscribe) {
        return new Observable<T>(onSubscribe);
    }

    public <R> Observable<R> map(Func1<T, R> function) {
        return new Observable<R>(new OnSubscribeMap<>(this, function));
    }

    public final Observable<T> observeOn(Scheduler scheduler) {
        return observeOn(scheduler);
    }

    public final Observable<T> observeOn(Scheduler scheduler, int size) {
        return lift(new OperatorObserveOn<T>(scheduler, size));
    }

    private <R> Observable<R> lift(final Operator<R, T> operator) {
        return new Observable<R>(new OnSubscribeLift<>(this.subscribe, operator));
    }

    @Override
    public Subscription subscribe(@NonNull Observer<T> observer) {
        return subscribe(new ObserverSubscriber<T>(observer));
    }

    public Subscription subscribe(@NonNull Subscriber<T> subscriber) {
        return subscribe(subscriber, this);
    }

    static <T> Subscription subscribe(Subscriber<T> subscriber, Observable<T> observable) {
        // validate and proceed
        if (subscriber == null) {
            throw new IllegalArgumentException("subscriber can not be null");
        }
        if (observable.subscribe == null) {
            throw new IllegalStateException("onSubscribe function can not be null.");
            /*
             * the subscribe function can also be overridden but generally that's not the appropriate approach
             * so I won't mention that in the exception
             */
        }

        // new Subscriber so onStart it
        subscriber.onStart();

        // The code below is exactly the same an unsafeSubscribe but not used because it would
        // add a significant depth to already huge call stacks.
        try {
            // allow the hook to intercept and/or decorate
            observable.subscribe.call(subscriber);
            return subscriber;
        } catch (Throwable e) {
            return Subscriptions.unsubscribed();
        }
    }
}
